package com.example.flinkcourse.lesson3.sink;

import com.example.flinkcourse.lesson3.model.Transaction;
import org.apache.flink.connector.hbase.sink.HBaseSink;
import org.apache.flink.streaming.api.functions.sink.Sink;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.HColumnDescriptor;
import org.apache.hadoop.hbase.client.HTableDescriptor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * HBase 输出工厂类
 * 用于创建 HBase 输出，支持 Kerberos 认证
 */
public class HBaseSinkFactory {
    private static final Logger LOG = LoggerFactory.getLogger(HBaseSinkFactory.class);

    /**
     * 创建 HBase 输出
     * @return HBase 输出
     */
    public static Sink<Transaction> createHBaseSink() {
        LOG.info("Creating HBase sink...");
        
        // 配置 HBase 连接
        Configuration configuration = HBaseConfiguration.create();
        configuration.set("hbase.zookeeper.quorum", "localhost:2181");
        
        // 配置 Kerberos 认证
        if (isKerberosEnabled()) {
            LOG.info("Kerberos authentication is enabled");
            configuration.set("hbase.security.authentication", "kerberos");
            configuration.set("hbase.kerberos.principal", "hbase/host@REALM");
            configuration.set("hbase.kerberos.keytab", "/path/to/keytab");
        }
        
        // 创建 HBase 表描述符
        TableName tableName = TableName.valueOf("output-table");
        HTableDescriptor tableDescriptor = new HTableDescriptor(tableName);
        tableDescriptor.addFamily(new HColumnDescriptor("cf"));
        
        // 创建 HBase 输出
        HBaseSink<Transaction> sink = HBaseSink.<Transaction>builder()
            .setTableName("output-table")
            .setHBaseConfiguration(configuration)
            .setSerializationSchema(new TransactionSerializationSchema())
            .build();
        
        LOG.info("HBase sink created successfully");
        return sink;
    }
    
    /**
     * 检查是否启用 Kerberos 认证
     * @return 是否启用 Kerberos
     */
    private static boolean isKerberosEnabled() {
        // TODO: 从配置中读取是否启用 Kerberos
        return true;
    }
}

/**
 * 交易序列化模式
 * 用于将交易数据序列化为 HBase 写入格式
 */
class TransactionSerializationSchema implements SerializationSchema<Transaction> {
    @Override
    public byte[] serialize(Transaction transaction) {
        // 将交易数据序列化为字节数组
        // 这里使用简单的 JSON 序列化作为示例
        try {
            ObjectMapper mapper = new ObjectMapper();
            return mapper.writeValueAsBytes(transaction);
        } catch (Exception e) {
            LOG.error("Error serializing transaction: " + transaction.getId(), e);
            return null;
        }
    }
} 